/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package edu.isi.pegasus.planner.subworkflow.scheduler;

import edu.isi.pegasus.planner.classes.ADag;
import edu.isi.pegasus.planner.classes.PCRelation;
import edu.isi.pegasus.planner.dax.ADAG;
import edu.isi.pegasus.planner.dax.ParADAG;
import edu.isi.pegasus.planner.subworkflow.partitioner.DAXAbstraction;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Vector;

/**
 *
 * @author wchen
 *
 **/

  class writeToFile implements Runnable
{
     private ADAG dax;
     Thread runner;
     private boolean done;
     private String daxPath;
     public void getDone()
     {
         while(!done){
             try
             {
                Thread.sleep(10000);
             }
             catch (Exception e)
             {
                 e.printStackTrace();
             }

         }
     }
     public writeToFile(ADAG dax, String daxPath)
     {
         runner = new  Thread(this, "");
        this.dax = dax;
        this.daxPath = daxPath;
        this.done   = false;
        runner.start();
     }
    public void run()
    {
        dax.writeToFile(daxPath);
        this.done = true;
    }
}

class checkDAXRel implements Runnable{

    private ArrayList parentDAXList;
    private ArrayList childDAXList;
    private boolean done;
    private ADag mDag;
    private Map jobMap;
    private Map daxMap;
    private Map ID2LogicalID;
    private int start;
    private int end;
    Thread runner;
    @SuppressWarnings("static-access")
    public void getDone()
    {
         while(!done){
             try
             {
                Thread.currentThread().sleep(1000);
             }
             catch (Exception e)
             {
                 e.printStackTrace();
             }

         }
    }
    public void addDAX(ArrayList pDAX, ArrayList cDAX)
    {
        for(int i = 0; i < parentDAXList.size(); i++)
        {
            String sName = (String)parentDAXList.get(i);
            String cName = (String)childDAXList.get(i);
            pDAX.add(sName);
            cDAX.add(cName);
        }
        //System.out.println(parentDAXList.size() +" " + childDAXList.size());
    }
    public checkDAXRel(int start, int end, ADag mDag, Map jobMap, Map daxMap, Map ID2LogicalID)
    {
        runner = new Thread(this, "chcke");
        this.mDag = mDag;
        this.parentDAXList = new ArrayList();
        this.childDAXList = new ArrayList();
        this.jobMap = jobMap;
        this.start = start;
        this.end = end;
        this.daxMap = daxMap;
        this.ID2LogicalID = ID2LogicalID;
        done = false;
        runner.start();
        
    }
    @Override
    public void run()
    {
        String pFake="";
        String cFake="";
        Map dag2bigDax = new HashMap<ADAG, DAXAbstraction>();
        for(Iterator it = daxMap.keySet().iterator();it.hasNext();)
        {
            DAXAbstraction bigDax = (DAXAbstraction)it.next();
            dag2bigDax.put(bigDax.adag, bigDax);
        }


            for( int i = start; i < end && i < mDag.vJobSubInfos.size(); i++ ){
                edu.isi.pegasus.planner.classes.Job dagJob =
                        (edu.isi.pegasus.planner.classes.Job)mDag.vJobSubInfos.get(i);
               Vector parentList = mDag.getParents(dagJob.getID());

               for(Iterator pIt = parentList.iterator();pIt.hasNext();)
               {
                   String parent = (String)pIt.next();
 //                  String parentName = mDag.getSubInfo(parent).getLogicalID();
                   String parentName = (String)ID2LogicalID.get(parent);
                   String childName = dagJob.getLogicalID();
                   ParADAG childDAX = (ParADAG)jobMap.get(childName);
                   ParADAG parentDAX = (ParADAG)jobMap.get(parentName);
//                   ADAG childDAX = (ADAG)jobMap.get(dagJob.jobID);
//                   ADAG parentDAX = (ADAG)jobMap.get(parent);
                   if(childDAX.equals(parentDAX))
                   {
                       childDAX.addDependency(parentName, childName);
                       DAXAbstraction bigDax = (DAXAbstraction)dag2bigDax.get(childDAX);
                       synchronized(bigDax.subDag)
                       {
                           bigDax.subDag.addNewRelation(parent, dagJob.jobID);
                       }
                       

                   }
                   else
                   {

                       if(!pFake.equals(parentDAX.getName())||!cFake.equals(childDAX.getName()))
                        {
                           synchronized(parentDAXList)
                           {
                               synchronized (childDAXList)
                               {
                                    parentDAXList.add(parentDAX.getName());
                                    childDAXList.add(childDAX.getName());
                               }


                           }

                            pFake = parentDAX.getName();
                            cFake = childDAX.getName();
                       }


                   }

               }

            }
            done = true;
    }
}

class checkDAXRel2 implements Runnable{

    private ArrayList parentDAXList;
    private ArrayList childDAXList;
    private boolean done;
    private ADag mDag;
    private Map jobMap;
    private Map daxMap;
    private Map ID2LogicalID;
    private int index;
    private int num2slot;
    Thread runner;
    @SuppressWarnings("static-access")
    public void getDone()
    {
         while(!done){
             try
             {
                Thread.currentThread().sleep(1000);
             }
             catch (Exception e)
             {
                 e.printStackTrace();
             }

         }
    }
    public void addDAX(ArrayList pDAX, ArrayList cDAX)
    {
        for(int i = 0; i < parentDAXList.size(); i++)
        {
            String sName = (String)parentDAXList.get(i);
            String cName = (String)childDAXList.get(i);
            pDAX.add(sName);
            cDAX.add(cName);
        }
        //System.out.println(parentDAXList.size() +" " + childDAXList.size());
    }
    public checkDAXRel2(int index, int num2slot, ADag mDag, Map jobMap, Map daxMap, Map ID2LogicalID)
    {
        runner = new Thread(this, "chcke");
        this.mDag = mDag;
        this.parentDAXList = new ArrayList();
        this.childDAXList = new ArrayList();
        this.jobMap = jobMap;
        this.index  = index;
        this.num2slot = num2slot;
        this.daxMap = daxMap;
        this.ID2LogicalID = ID2LogicalID;
        done = false;
        runner.start();

    }
    @Override
    public void run()
    {
        String pFake="";
        String cFake="";
        Map dag2bigDax = new HashMap<ADAG, DAXAbstraction>();
        for(Iterator it = daxMap.keySet().iterator();it.hasNext();)
        {
            DAXAbstraction bigDax = (DAXAbstraction)it.next();
            dag2bigDax.put(bigDax.adag, bigDax);
        }


        Vector relation = mDag.dagInfo.relations;
        int size = relation.size();

        for(int i = index; i < size; i += num2slot)
        {
            PCRelation rel = (PCRelation)relation.get(i);
            String parent = rel.parent;
            String child  = rel.child;
            String parentName = (String)ID2LogicalID.get(parent);
            String childName  = (String)ID2LogicalID.get(child);
            ParADAG childDAX = (ParADAG)jobMap.get(childName);
            ParADAG parentDAX = (ParADAG)jobMap.get(parentName);

            if(childDAX.equals(parentDAX))
                   {
                        DAXAbstraction bigDax = (DAXAbstraction)dag2bigDax.get(childDAX);
                       synchronized(bigDax.subDag)
                       {
                            childDAX.addDependency(parentName, childName);
                            
                            bigDax.subDag.addNewRelation(parent, child);
                       }


                   }
                   else
                   {

                       if(!pFake.equals(parentDAX.getName())||!cFake.equals(childDAX.getName()))
                        {
                           synchronized(parentDAXList)
                           {
                               synchronized (childDAXList)
                               {
                                    parentDAXList.add(parentDAX.getName());
                                    childDAXList.add(childDAX.getName());
                               }


                           }

                            pFake = parentDAX.getName();
                            cFake = childDAX.getName();
                       }


                   }

        }

        done = true;
    }
}


